
package remote;

import java.io.Serializable;
import java.rmi.AccessException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import rebalance.RebalanceChecks;

import node.NodeManager;
import node.NodeRemoval;

import log.Logger;
import messaging.HashTableMessage;
import messaging.InterNodeMessage;
import messaging.MessageHashing;
import messaging.TokenPacket;
import messaging.TokenType;

/**
 * UnicastRemoteDistributedHashMap that implements RemoteDistributedMap( the remote interface)
 * Implements the remote interface methods.
 * 
 * @author Team-4
 * @version Spring 2013
 */
public class UnicastRemoteDistributedHashMap implements RemoteDistributedHashMap
{
  /**
   * The HashMap encapsulated by this RemoteDistributedHashtable.
   */
  public Map<String, Serializable> my_map = new HashMap<String, Serializable>();
  /**
   * hash map of all stubs to update it when the answer arrives.
   */
  public HashMap<Integer, FutureMessage> my_futures = new HashMap<Integer, FutureMessage>();
  /**
   * Stores the identity for get,put and remove methods. Get = 1 Put = 2 Remove
   * = 3
   */
  private int PUT_KEY = 1;
  /**
   * Get key.
   */
  private int GET_KEY = 2;
  /**
   * remove key.
   */
  private int REMOVE_KEY = 3;
  /**
   * Multiple store.
   */
  private int MULTIPLESTORE = 8;
  /**
   * wait time to send message to the rebalanced node.
   */
  private int WAIT_MESSAGE = 1000;
  /**
   * Stores the node number of this remote object.
   */
  private final int my_node_number;
  /**
   * Stores whether this node is real/actual.
   */
  private final boolean actual_virtual;
  /**
   * Stores the node number that has simulated this node.If this is a real node
   * it will store its own number.
   */
  private final int my_simulating_node;
  /**
   * Stores an instance of NodeManager
   */
  private NodeManager nodemanager;
  /**
   * A unique id generator for storing the futures.
   */
  private int uniquefutureid;
  /**
   * Is the message is done.
   */
  private final Object my_message_monitor = new Object();
  /**
   * Creates the Node Object for each registry object.
   * 
   * @param the_node_no : stores the node number
   */
  public UnicastRemoteDistributedHashMap(final int the_node_no, 
                                         final boolean act_vir,
                                         final int simulating_node,NodeManager nm)
  {
    my_node_number = the_node_no;
    actual_virtual = act_vir;
    my_simulating_node = simulating_node;
    this.nodemanager = nm;
    this.uniquefutureid = 0;
  }
  /**
   * This method shows if the node is available to take messages after re-balancing.
   * @return boolean
   */
  public boolean isrebalancing()
  {
    return this.nodemanager.rebalancing;
  }
  /**
   * This method generates a unique future id to store all the futures.
   * @return int - the unique id.
   */
  public synchronized int generateuniqueid()
  {
    return this.uniquefutureid++;
  }
  /**
   * Returns the data hash map of the data.
   * @return the hashmap data
   */
  public Map<String, Serializable> getData()
  {
    return my_map;
  }
  /**
   * set hash map data of this node.
   */
  public void setData(final Map<String, Serializable> my_data)
  {
    this.my_map = my_data;
  }
  /**
   * @return int - the node number which is simulating this node.
   */
  public int getSimulatingNode()
  {
    return my_simulating_node;
  }
  /**
   * @return the node number of this node
   */
  public int getNodeNumber()
  {
    return my_node_number;
  }
  /**
   * @return whether this is a real or a virtual node
   */
  public boolean getNodeStatus()
  {
    return actual_virtual;
  }
  /**
   * The get method implemented.
   * Given a key , this will return the value associated with this key.
   * @param the_key 
   * @return the future returned to client.
   */
  @Override
  public RemoteFuture get(final String the_key) 
  {
    return this.compute(the_key, null, GET_KEY);
  }
  /**
   * The put method implemented.
   * @param the_key 
   * @param the_value
   * @return the future returned to client.
   */
  @Override
  public RemoteFuture put(final String the_key, final Serializable the_value) 
  {
    return this.compute(the_key, the_value, PUT_KEY);
  }
  /**
   * The remove method implemented.
   * @param the_key
   * @return the future returned to client.
   */
  @Override
  public RemoteFuture remove(final String the_key) 
  {
    return this.compute(the_key, null, REMOVE_KEY);
  }
  @Override
  /**
   * The send method to put the messages on the queue.
   * @param my_message - the inter-node message
   */
  public void send(final InterNodeMessage my_message)
  {
    try
    {
      nodemanager.node_channel.get(this.my_node_number)
          .getChannelQueue(my_message.getMy_source_node()).getQueue().put(my_message);
    }
    catch (final InterruptedException e)
    {
      System.out.println("thread interrupted when waiting ");
      System.exit(1);
    }
  }
  /**
   * this method is the landing point for all method calls on this remote
   * object. It will do the hashing and routing.
   * @param the_key - the key value
   * @param the_value - the value
   * @param method_name - the method name( get,put, remove)
   * @return the future to be returned to client.
   */
  public RemoteFuture compute(final String the_key, final Serializable the_value, final int method_name)
  {
    int dest_node = 0;
    try
    {
      Logger.out(" A rmi call received from client for key " + the_key + 
                 " in node " + this.my_node_number);
      final int forward_node = MessageHashing.findHashValue(the_key);
      final FutureMessage ex = new FutureMessage(the_key);
      final int uniquefutureid = this.generateuniqueid();
      my_futures.put(uniquefutureid, ex);
      if (forward_node == this.my_node_number)
      {
        Logger.out( this.my_node_number + " The key " + the_key + " hashes to the same node ");
        Serializable val = null;
        if (method_name == PUT_KEY)
        {
          val = this.my_map.put(the_key, the_value);
          //put data in the predecessor actual node too.
          this.nodemanager.duplicatestores.sendMessagePredecessor(the_key, the_value, 
                                                                  forward_node, PUT_KEY, 
                                                                  this.my_node_number);
          ex.set(val);
          my_futures.remove(uniquefutureid);
          Logger.out(this.my_node_number + " The future for " + the_key + " is updated and sent ");
        }
        if (method_name == GET_KEY)
        {
          val = this.my_map.get(the_key);
          ex.set(val);
          my_futures.remove(uniquefutureid);
          Logger.out(this.my_node_number + " The future for " + the_key + " is updated and sent ");
        }
        if (method_name == REMOVE_KEY)
        {
          val = this.my_map.remove(the_key);
          this.nodemanager.duplicatestores.sendMessagePredecessor(the_key, the_value, 
                                                                  forward_node, REMOVE_KEY, 
                                                                  this.my_node_number);
          ex.set(val);
          my_futures.remove(uniquefutureid);
          Logger.out(this.my_node_number + " The future for " + the_key + " is updated and sent ");
        }
        final RemoteFuture stub = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
        return stub;
      }
      else
      {
        dest_node = nodemanager.hashhelp.optimalNodeFind(this.my_node_number, forward_node);
        Logger.out(this.my_node_number + "  the key " + the_key + " hashes to the node - " + forward_node);
        final RemoteFuture stub1 = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
        InterNodeMessage message = null;
        HashTableMessage htm = null;
        if (method_name == GET_KEY)
        {
          message = new InterNodeMessage(this.my_node_number, forward_node, 
                                         this.my_node_number, dest_node, GET_KEY);
          message.setAnswer_set(false);
          htm = new HashTableMessage(the_key, null, forward_node);
          htm.setUniqueid(uniquefutureid);
          message.setHashTableMessage(htm);
        }
        if (method_name == PUT_KEY)
        {
          message = new InterNodeMessage(this.my_node_number, forward_node, 
                                         this.my_node_number, dest_node, PUT_KEY);
          message.setAnswer_set(false);
          htm = new HashTableMessage(the_key, the_value, forward_node);
          htm.setUniqueid(uniquefutureid);
          message.setHashTableMessage(htm);
        }
        if (method_name == REMOVE_KEY)
        {
          message = new InterNodeMessage(this.my_node_number, forward_node, 
                                         this.my_node_number, dest_node, REMOVE_KEY);
          message.setAnswer_set(false);
          htm = new HashTableMessage(the_key, null, forward_node);
          htm.setUniqueid(uniquefutureid);
          message.setHashTableMessage(htm);
        }
        final String dest_node_str = nodemanager.rmi_NodeFind(dest_node);
        try
        {
          final RemoteDistributedHashMap to_send =
            (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
          Logger.out(this.my_node_number + " Sending message- " + the_key + 
                     " to node- " + dest_node);
          to_send.send(message);
          Logger.out(this.my_node_number + " message - " + the_key + " sent");
        }
        catch (final RemoteException e)
        {
          Logger.out("unable to send message - " + the_key +
                     " in node " + this.my_node_number);
          RebalanceChecks reb = 
              new RebalanceChecks(this.nodemanager, this.my_node_number, dest_node);
          int predecessor = reb.findPredecessor(dest_node);
          if (this.nodemanager.rmi_NodeFind(dest_node).endsWith("V"))
          {
            predecessor = reb.findPredecessor(predecessor);
          }
          Logger.out(" the pred is " + predecessor);
          reb.doallchecks();
          handlemessage(message,predecessor);
        }
        return stub1;
      }
    }
    catch (final NotBoundException e)
    {
      System.out.println(" the registry name is not bound ");
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      System.out.println("unable to access the registry in unicast");
      System.exit(1);
    }
    catch (final InterruptedException e)
    {
      System.out.println(" interrupted during thread wait");
      System.exit(1);
    }
    return null;
  }
  /**
   * This method handles the message that caused to throw re-balancing.
   * @throws NotBoundException 
   * @throws RemoteException 
   * @throws AccessException 
   * @throws InterruptedException 
   */
  public void handlemessage(InterNodeMessage message ,int predecessor) 
      throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    
    final long time = System.currentTimeMillis();
    long newtime = System.currentTimeMillis();
    while (newtime - time < WAIT_MESSAGE)
    {
      newtime = System.currentTimeMillis();
    }
    String dest_node_str = nodemanager.rmi_NodeFind(predecessor);
    RemoteDistributedHashMap to_send =
        (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    while (to_send.isrebalancing())
    {
      //Logger.out(" rebalancing");
    }
    dest_node_str = nodemanager.rmi_NodeFind(message.getMy_dest_node());
    to_send = (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.out(this.my_node_number + " Sending message- " +
          message.getHashTableMessage().getkey() + " to node- " + dest_node_str);
    to_send.send(message);
  }
  @Override
  public void removeNode() 
      throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    NodeRemoval noderem = new NodeRemoval(this.my_node_number, NodeManager.total_nodes_chord, 
                                          NodeManager.registry, 
                                          nodemanager.real_virtual_node_info,this.nodemanager);
    noderem.removeNode();
  }
  
  public RemoteFuture firstKey() throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Got a client request for first key in ring");
    TokenPacket token_packet = new TokenPacket(this.my_node_number, 
                                               TokenType.FIRSTKEY);  
    final Map<String, Serializable> firstkey = 
        nodemanager.node_stubs.get(this.my_node_number).my_map;
    final Set<String> string_value = firstkey.keySet(); 
    //final int forward_node = MessageHashing.findHashValue("FIRST");
    final FutureMessage ex = new FutureMessage("FIRST");
    final int uniquefutureid = this.generateuniqueid();
    my_futures.put(uniquefutureid, ex);
    final RemoteFuture stub1 = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
    if(!string_value.isEmpty())
    {
      final Iterator i = string_value.iterator();
      String first_value = i.next().toString();
      while (i.hasNext())
      {
        final String set_string = i.next().toString();
        if (set_string.compareTo(first_value) < 0)
        {
          first_value = set_string; 
        }
      }
      token_packet.setMyFirstKey(first_value);
     
    }
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Creating the first key token");
    //token_packet.setMyNodeCount(1);
    final InterNodeMessage inter_message = new InterNodeMessage(this.my_node_number, 
                                                                this.my_node_number + 1, 
                                                                this.my_node_number, 
                                                                this.my_node_number + 1, 7);
    HashTableMessage htm = new HashTableMessage("FIRST", null, this.my_node_number + 1);
    htm.setUniqueid(uniquefutureid);
    inter_message.setHashTableMessage(htm);
    inter_message.setAnswer_set(false);
    inter_message.setMyTokenPacket(token_packet);
    final String dest_node_str;
    if (this.my_node_number + 1 >= 8)
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(0);
    }
    else 
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(this.my_node_number + 1);
    }
    final RemoteDistributedHashMap to_send =
         (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Sending token message to the next node");
    to_send.send(inter_message);
    return stub1;
  }
  
  
  public RemoteFuture lastKey() throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Got a client request for last key in ring");
    TokenPacket token_packet = new TokenPacket(this.my_node_number, 
                                               TokenType.LASTKEY);  
    final Map<String, Serializable> lastkey = 
        nodemanager.node_stubs.get(this.my_node_number).my_map;
    //final int forward_node = MessageHashing.findHashValue("FIRST");
    final FutureMessage ex = new FutureMessage("LAST");
    final int uniquefutureid = this.generateuniqueid();
    final RemoteFuture stub1 = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
    my_futures.put(uniquefutureid, ex);
    final Set<String> string_value = lastkey.keySet(); 
    if (!string_value.isEmpty())
    {
      final Iterator i = string_value.iterator();
      String last_value = i.next().toString();
      while (i.hasNext())
      {
        final String set_string = i.next().toString();
        if (set_string.compareTo(last_value) > 0)
        {
          last_value = set_string; 
        }
      }
      token_packet.setMyLastKey(last_value);
    }
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Token message is creating the last key token");
    //token_packet.setMyNodeCount(1);
    final InterNodeMessage inter_message = new InterNodeMessage(this.my_node_number, 
                                                                this.my_node_number + 1, 
                                                                this.my_node_number, 
                                                                this.my_node_number + 1, 7);
    HashTableMessage htm = new HashTableMessage("LAST", null, this.my_node_number + 1);
    htm.setUniqueid(uniquefutureid);
    inter_message.setHashTableMessage(htm);
    
    inter_message.setAnswer_set(false);
    inter_message.setMyTokenPacket(token_packet);
    final String dest_node_str;
    if (this.my_node_number + 1 >= 8)
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(0);
    }
    else 
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(this.my_node_number + 1);
    }
    
    final RemoteDistributedHashMap to_send =
         (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Sending token message to the next node");
    to_send.send(inter_message);
    return stub1;
  }
  
  
  public RemoteFuture totalCount() throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Got a client request for total count of keys in ring");
    TokenPacket token_packet = new TokenPacket(this.my_node_number, 
                                               TokenType.COUNT);  
    
    int count = 0;
    final Map<String, Serializable> firstkey = 
        nodemanager.node_stubs.get(this.my_node_number).my_map;
    final Set<String> string_value = firstkey.keySet(); 
    count += string_value.size(); 
    token_packet.setMyKeyCount(count);
    final int forward_node = MessageHashing.findHashValue("COUNT");
    final FutureMessage ex = new FutureMessage("COUNT");
    final int uniquefutureid = this.generateuniqueid();
    final RemoteFuture stub1 = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
    my_futures.put(uniquefutureid, ex);
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Creating the total node count token");
    final InterNodeMessage inter_message = new InterNodeMessage(this.my_node_number, 
                                                                this.my_node_number + 1, 
                                                                this.my_node_number, 
                                                                this.my_node_number + 1, 7);
    HashTableMessage htm = new HashTableMessage("FIRST", null, this.my_node_number + 1);
    htm.setUniqueid(uniquefutureid);
    inter_message.setHashTableMessage(htm);
    inter_message.setAnswer_set(false);
    inter_message.setMyTokenPacket(token_packet);
    final String dest_node_str;
    if (this.my_node_number + 1 >= 8)
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(0);
    }
    else 
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(this.my_node_number + 1);
    }
    final RemoteDistributedHashMap to_send =
         (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Sending token message to the next node");
    to_send.send(inter_message);
    return stub1; 
  }
  
  
  public RemoteFuture actualCount() throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Got a request for actual node list in ring");
    TokenPacket token_packet = new TokenPacket(this.my_node_number, 
                                               TokenType.ACTUALCOUNT);  
    //final int forward_node = MessageHashing.findHashValue("FIRST");
    final FutureMessage ex = new FutureMessage("ACTUALCOUNT");
    final int uniquefutureid = this.generateuniqueid();
    final RemoteFuture stub1 = (RemoteFuture) UnicastRemoteObject.exportObject(ex, 0);
    my_futures.put(uniquefutureid, ex);
    if(nodemanager.actual_node_number == this.my_node_number)
    {
      ArrayList<Integer> list = new ArrayList<>();
      list.add(this.my_node_number);
      token_packet.setMyActualCount(list);
    }
    else 
    {
      token_packet.setMyActualCount(new ArrayList<Integer>());
    }
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Creating the actual node token");
    final InterNodeMessage inter_message = new InterNodeMessage(this.my_node_number, 
                                                                this.my_node_number + 1, 
                                                                this.my_node_number, 
                                                                this.my_node_number + 1, 7);
    inter_message.setAnswer_set(false);
    HashTableMessage htm = new HashTableMessage("FIRST", null, this.my_node_number + 1);
    htm.setUniqueid(uniquefutureid);
    inter_message.setHashTableMessage(htm);
    inter_message.setMyTokenPacket(token_packet);
    final String dest_node_str;
    if (this.my_node_number + 1 >= 8)
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(0);
    }
    else 
    {
      dest_node_str = this.nodemanager.rmi_NodeFind(this.my_node_number + 1);
    }
    final RemoteDistributedHashMap to_send =
        (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.writeContent("LOG-"+this.my_node_number  + ".txt", "Sending token message to the next node");
    to_send.send(inter_message);
    return stub1;
  }
  
}
